package com.shujia.flink.window;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class Demo01TimeWindow {
    public static void main(String[] args) throws Exception {
        /*
         * 1、基于时间的窗口
         *       基于事件时间：
         *              滑动：SlidingEventTimeWindows
         *              滚动：TumblingEventTimeWindows
         *       基于处理时间：
         *              滑动：SlidingProcessingTimeWindows
         *              滚动：TumblingProcessingTimeWindows
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> wordDS = env.socketTextStream("master", 8888);

        // 如果后续需要基于事件时间进行统计 则需要设置事件时间以及水位线
        DataStream<Tuple2<String, Long>> assignDS = wordDS.map(line -> {
                    String[] split = line.split(",");
                    String word = split[0];
                    long ts = Long.parseLong(split[1]);
                    return Tuple2.of(word, ts);
                }, Types.TUPLE(Types.STRING, Types.LONG))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Tuple2<String, Long>>forMonotonousTimestamps()
                        .withTimestampAssigner((kv, ts) -> kv.f1)
                );

        // 每个5s统计最近10s内的单词数量
        assignDS.map(kv -> Tuple2.of(kv.f0, 1),Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(kv -> kv.f0)
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .sum(1)
                .print();

        env.execute();

    }
}
